package com.smsc.headend.task.engine.handler.concurrent;

import cn.hutool.extra.spring.SpringUtil;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.task.engine.executor.concurrent.CheckTaskExecutorRunner;
import com.smsc.headend.task.engine.service.task.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;

import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.*;


@Slf4j
public class TaskRejectHandler implements RejectedExecutionHandler {
    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @Autowired
    TaskService taskService;


    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        Collection<MessageListenerContainer> allListenerContainers = kafkaListenerEndpointRegistry.getAllListenerContainers();
        MessageListenerContainer listenerContainer = allListenerContainers.stream()
                .filter(listener -> Arrays.stream(listener.getContainerProperties().getTopics()).anyMatch(s -> TaskKafkaTopic.TASK_SEND_TO_UTE.equals(s))).findFirst().get();
        if (listenerContainer.isPauseRequested()) {
            try {
                log.warn("ThreadPool {} Full,  waiting For pause", executor.getThreadFactory().toString());
                Thread.sleep(2000);
            } catch (InterruptedException e) {

            }
        } else {
            log.warn("ThreadPool {} Full, Stop ConsumeTemporarily", executor.getThreadFactory().toString());
            listenerContainer.pause();
        }
        taskService.saveRejectedTask(r);
        taskService.scheduleCheckExecutor();
    }


}
